/*
 * Copyright ©2021 su binglun(9085309@qq.com). All rights reserved.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 *
 *---------------------------------------------------------------------
 *-                                                                   -
 *-           Website : http://phoenix.sunny360.cn/                   -
 *-                                                                   -
 *---------------------------------------------------------------------
 */
package cn.sunny360.core.handler.eb;

import cn.sunny360.core.service.UserService;
import cn.sunny360.core.utils.Container;
import cn.sunny360.messsage.MsgMessage;
import io.vertx.core.Promise;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.MessageConsumer;
import io.vertx.core.http.ServerWebSocket;
import io.vertx.core.impl.logging.Logger;
import io.vertx.core.impl.logging.LoggerFactory;

public class EbConsumerHandler {
    private static Logger logger = LoggerFactory.getLogger(EbConsumerHandler.class);

    private UserService userService;

    public EbConsumerHandler() {
        userService = Container.getComponent(UserService.class);
    }
    public void eventBusConsumer(EventBus eventBus, ServerWebSocket webSocket, String token) {
        //注册消费者
        MessageConsumer<Object> consumer = eventBus.consumer(token);
        consumer.handler(msg->{
            //接收到消息
            Buffer buffer = (Buffer) msg.body();
            //取消息内容
            MsgMessage eMsg = MsgMessage.newInstance();
            eMsg.parseByteArray(buffer.getBytes());
            String type = eMsg.getType();
            if ("close".equals(type)) {
                //连接断开
                msgCloseHandler(consumer, token, webSocket);
            } else {
                msgMessageHandler(eMsg, webSocket);
            }
        });
    }

    private void msgCloseHandler(MessageConsumer<?> consumer, String token, ServerWebSocket webSocket) {
        if (!webSocket.isClosed()) {
            webSocket.close();
        }
        //从在线用户列表中删除
        Promise<String> promise = Promise.promise();
        userService.wsDisconnection(token, promise);
        promise.future()
                .onSuccess(result->{
                    logger.debug("删除在线用户列表:" + result);
                })
                .onFailure(e->{
                    logger.error(e);
                });
        logger.debug("注销EventBus的Consumer");
        //连接关闭
        consumer.unregister();
    }


    private void msgMessageHandler(MsgMessage eMsg, ServerWebSocket webSocket) {
        //将消息发送给客户端
        webSocket.write(Buffer.buffer(eMsg.toByteArray()));
    }
}
